package com.atguigu.springdgademo.meta.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.support.spring.PropertyPreFilters;
import com.atguigu.springdgademo.meta.bean.TableMetaInfo;
import com.atguigu.springdgademo.meta.mapper.TableMetaInfoMapper;
import com.atguigu.springdgademo.meta.service.TableMetaInfoExtraService;
import com.atguigu.springdgademo.meta.service.TableMetaInfoService;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.apache.calcite.avatica.Meta;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.thrift.TException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * <p>
 * 元数据表 服务实现类
 * </p>
 *
 * @author wumingshi
 * @since 2024-08-29
 */
@Service
@DS("mydga")
public class TableMetaInfoServiceImpl extends ServiceImpl<TableMetaInfoMapper, TableMetaInfo> implements TableMetaInfoService {

    @Autowired
    TableMetaInfoExtraService tableMetaInfoExtraService;


    @Override
    public void initTableMetaInfo(String schemaName, String assessDate) {
        //提取元数据信息
        //获取表
        try {
            //清除当日提取的元数据
            this.remove(
              new QueryWrapper<TableMetaInfo>()
                        .eq("assess_date", assessDate)
            );

            List<String> allTableNames = hiveClient.getAllTables(schemaName);
            //System.out.println("allTableNames = " + allTableNames);

            //创建集合，存储批次数据
            ArrayList<TableMetaInfo> tableMetaInfos = new ArrayList<>(allTableNames.size());
            //获取表对象
            for (String tableName : allTableNames) {
                Table table = hiveClient.getTable(schemaName, tableName);
                //System.out.println(table);

                //封装
                TableMetaInfo tableMetaInfo = extractTableMetaInfoFromHive(table);
                //System.out.println(tableMetaInfo);

                //hdfs的元数据
                extractTableMetaInfoFromHdfs(tableMetaInfo);
                //System.out.println("tableMetaInfo = " + tableMetaInfo);

                //补充其他信息
                tableMetaInfo.setAssessDate(assessDate);
                tableMetaInfo.setCreateTime(new Date());

                //赞批次
                tableMetaInfos.add(tableMetaInfo);
            }
            //批次写入数据库
            this.saveBatch(tableMetaInfos);

            //初始化表的辅助信息
            tableMetaInfoExtraService.initTableMetaInfoExtra(tableMetaInfos);

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private void extractTableMetaInfoFromHdfs(TableMetaInfo tableMetaInfo) throws Exception {
        //创建文件系统对象
        FileSystem fs = FileSystem.get(
          new URI(tableMetaInfo.getTableFsPath()),
          new Configuration(),
          tableMetaInfo.getTableFsOwner()
        );

        //获取当前对象在hdfs中的所有内容
        FileStatus[] fileStatuses = fs.listStatus(new Path(tableMetaInfo.getTableFsPath()));

        //通过递归方式获取表大小
        addHdfsInfo(fs,fileStatuses,tableMetaInfo);

        //hdfs文件系统的容量信息
        tableMetaInfo.setFsCapcitySize(fs.getStatus().getCapacity());
        tableMetaInfo.setFsUsedSize(fs.getStatus().getUsed());
        tableMetaInfo.setFsRemainSize(fs.getStatus().getRemaining());
    }

    private void addHdfsInfo(FileSystem fs, FileStatus[] fileStatuses, TableMetaInfo tableMetaInfo) throws IOException {
        for (FileStatus fileStatus : fileStatuses) {
            //判断是否文件
            if(fileStatus.isFile()){
                //是计算大小
                //表大小
                tableMetaInfo.setTableSize(
                        (tableMetaInfo.getTableSize() == null ? 0L : tableMetaInfo.getTableSize() + fileStatus.getLen())
                );

                //表总大小
                tableMetaInfo.setTableTotalSize(
                        (tableMetaInfo.getTableTotalSize() == null ? 0L : tableMetaInfo.getTableTotalSize() + fileStatus.getLen())
                );

                //最后访问时间
                long fileLastAccessTime = fileStatus.getAccessTime();
                long currLastAccessTime = tableMetaInfo.getTableLastAccessTime() == null ? 0L : tableMetaInfo.getTableLastAccessTime().getTime();
                long lastAccessTime = Math.max(fileLastAccessTime,currLastAccessTime);
                tableMetaInfo.setTableLastAccessTime(new Date(lastAccessTime));

                //最后修改时间
                long fileLastModifyTime = fileStatus.getModificationTime();
                long currLastModifyTime = tableMetaInfo.getTableLastModifyTime() == null ? 0L : tableMetaInfo.getTableLastModifyTime().getTime();
                long lastModifyTime = Math.max(fileLastModifyTime,currLastModifyTime);
                tableMetaInfo.setTableLastModifyTime(new Date(lastModifyTime));
            }else {
                FileStatus[] subFileStatuses = fs.listStatus(fileStatus.getPath());
                addHdfsInfo(fs,subFileStatuses,tableMetaInfo);
            }
        }

    }

    private TableMetaInfo extractTableMetaInfoFromHive(Table table) {

        TableMetaInfo tableMetaInfo = new TableMetaInfo();
        //获取后设置
        //库名
        tableMetaInfo.setSchemaName(table.getDbName());
        //表名
        tableMetaInfo.setTableName(table.getTableName());

        //列信息
        PropertyPreFilters propertyPreFilters = new PropertyPreFilters();
        PropertyPreFilters.MySimplePropertyPreFilter filter = propertyPreFilters.addFilter("comment","name","type");
        tableMetaInfo.setColNameJson(JSON.toJSONString(table.getSd().getCols(),filter));

        //分区列信息
        tableMetaInfo.setPartitionColNameJson(JSON.toJSONString(table.getPartitionKeys(),filter));

        //hdfs所属人
        tableMetaInfo.setTableFsOwner(table.getOwner());

        //表参数
        tableMetaInfo.setTableParametersJson(JSON.toJSONString(table.getParameters()));

        //表描述
        tableMetaInfo.setTableComment(table.getParameters().get("comment"));

        //表路径
        tableMetaInfo.setTableFsPath(table.getSd().getLocation());

        //表输入格式
        tableMetaInfo.setTableInputFormat(table.getSd().getInputFormat());

        //表输出格式
        tableMetaInfo.setTableOutputFormat(table.getSd().getOutputFormat());

        //行格式
        tableMetaInfo.setTableRowFormatSerde(table.getSd().getSerdeInfo().getSerializationLib());

        //表创建时间
        tableMetaInfo.setTableCreateTime(new Date( table.getCreateTime() * 1000L ));

        //表类型
        tableMetaInfo.setTableType(table.getTableType());

        //表的分桶列
        tableMetaInfo.setTableBucketColsJson(JSON.toJSONString(table.getSd().getBucketCols()));

        //表的分桶数
        tableMetaInfo.setTableBucketNum( (long)table.getSd().getNumBuckets() );

        //表的分桶排序字段
        tableMetaInfo.setTableSortColsJson(JSON.toJSONString(table.getSd().getSortCols()));

        return tableMetaInfo;
    }


    private IMetaStoreClient hiveClient;

    /*
    * 创建hive客户端对象
    * */

    //读取配置文件信息
    @Value("${hive.metastore.uris}")
    String hiveMetaStoreUris = "thrift://hadoop102:9083";

    @PostConstruct //当前对象创建后，调用一次方法
    public void createHiveClient(){
        Configuration configuration = new Configuration();
        //configuration.set();
        MetastoreConf.setVar(configuration, MetastoreConf.ConfVars.THRIFT_URIS,hiveMetaStoreUris);

        try {
            hiveClient = new HiveMetaStoreClient(configuration);
            System.out.println("hiveClient = " + hiveClient);
        } catch (MetaException e) {
            e.printStackTrace();
            throw new RuntimeException("获取hive客户端对象失败");
        }

    }
}
